package com.atbeijing.app

import com.alibaba.fastjson.JSON
import com.atbeijing.bean.UserInfo
import com.atbeijing.constants.GmallConstants
import com.atbeijing.utils.MyKafkaUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
 * 消费kafka数据,得到用户信息并存入redis
 * 这个逻辑不太合理
 */
object UserInfoApp {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("UserInfoApp")
    val ssc = new StreamingContext(conf,Seconds(5))

    val userInfoDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_USER,ssc)

    userInfoDS.foreachRDD(rdd =>{
      rdd.foreachPartition(partition =>{
        val jedis: Jedis = new Jedis("hadoop202",6379)
        partition.foreach(record =>{
          val userInfoStr: String = record.value()
          val userInfo: UserInfo = JSON.parseObject(userInfoStr,classOf[UserInfo])
          val userInfoKey: String = "userInfo"+userInfo.id
          //value 为String类型
          jedis.set(userInfoKey,userInfoStr)
        })
        jedis.close()
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }


}
